/*
 * Copyright (C) 2008 CERNET Network Center 
 * dvping/dvmcast/hdvping/hdvmcast series 
 * 
 * Design and coding: 
 *  Xing Li <xing@cernet.edu.cn> 
 *	Congxiao Bao <congxiao@cernet.edu.cn>
 * 	Jinpeng Jiang <jjp02@mails.tsinghua.edu.cn>
 * 
 * Contributions:
 * 
 * Permission to use, copy, modify, and distribute this software for any
 * purpose with or without fee is hereby granted, provided that the above
 * copyright notice and this permission notice appear in all copies.
 */


#include "dvping.h"

DVPinger::DVPinger(const IPAddress& host, const IPAddress& remote_host, in_port_t snd_port, in_port_t rcv_port) 
	: prth((rtp_trivial_hdr_t*)packet), pRtpHdr((rtp_hdr_t*)packet),
	snd_pkt_seq(*(int64_t*)(packet + sizeof(rtp_hdr_t))), 
	rcv_pkt_seq(*(int64_t*)(rcvbuf + sizeof(rtp_hdr_t)))
{
	this->host = host;
	this->remote_host = remote_host;
	this->snd_port = snd_port;
	this->rcv_port = rcv_port;

	evtRecvThreadQuit.Reset();
	rcv_thread = NullThread;
	bQuitRecv = false;
}
DVPinger::~DVPinger()
{
	Cleanup();
}

bool DVPinger::Init()
{
	snd_sock = Socket::CreateUdpSender(host.AddressFamily());
	snd_sock.Block(false);
	snd_sock.SetSendBufferSize(1000000);
	rcv_sock = Socket::CreateUdpReceiver(host.AddressFamily(), rcv_port, false);
	rcv_sock.SetRecvBufferSize(10000000);

	memset(packet, 0, packet_size);

	prth->version = RTP_VERSION;
	prth->pt = DEFAULT_DVTS_RTP_PT;

	srand(time(NULL));
    pRtpHdr->seq = htons(rand());
	pRtpHdr->ssrc = htons(rand());
	pRtpHdr->ts = htons(rand());

	evtRecvThreadQuit.Reset();
	rcv_thread = NullThread;
	bQuitRecv = false;

	return snd_sock.IsValid() && rcv_sock.IsValid();
}

struct dv_ping_share_data
{
	dv_ping_result	dpr;
	int64_t			first_recv_time;
	int64_t			period;
	double			bandwidth;
	int64_t			pkt_seq_divider;
	int64_t			delay_hit_pkt_count;
	int64_t			delay_sum;
	int64_t			start_time;
};


void* DVPinger::RecvProc(void* param)
{
	if(param == NULL)
		return NULL;
	dv_ping_share_data& dpsd = *(dv_ping_share_data*)param;
	IPEndPoint rcvep;
	while(!bQuitRecv)
	{
		// recv
		int32_t bytes_recv = 0;
		while((bytes_recv = rcv_sock.ReceiveFrom(rcvbuf, packet_size, rcvep, 1000)) > 0)
		{
			if(rcvep.Address() != remote_host)
				continue;
			int64_t now = DateTime::Now().GetTickCount();
			if(dpsd.first_recv_time == 0)
			{
				dpsd.first_recv_time = now;
			}
			if(rcv_pkt_seq % dpsd.pkt_seq_divider == 0)
			{
				int64_t index = rcv_pkt_seq / dpsd.pkt_seq_divider;
				if(index >= 0 && index < 1000 && snd_pkt_time_cache[index] > 0)
				{
					dpsd.delay_sum += now - snd_pkt_time_cache[index];
					++dpsd.delay_hit_pkt_count;
				}
			}
			++ dpsd.dpr.packets_recv;
			dpsd.dpr.bytes_recv += bytes_recv;
		}
	}
	rcv_thread = NullThread;
	evtRecvThreadQuit.Signal();
	return NULL;
}

dv_ping_result DVPinger::Ping(int64_t period, double bandwidth)
{
	dv_ping_share_data dpsd;

	snd_pkt_seq = 0;
	memset(snd_pkt_time_cache, 0, sizeof(snd_pkt_time_cache));
	dpsd.period = period;
	dpsd.bandwidth = bandwidth;
	dpsd.pkt_seq_divider = (int64_t)(period * bandwidth / packet_size) / 1000;
	if(dpsd.pkt_seq_divider == 0)
	{
		dpsd.pkt_seq_divider = 1;
	}
	dpsd.delay_sum = 0;
	dpsd.delay_hit_pkt_count = 0;
	dpsd.first_recv_time = 0;

	if(!snd_sock.IsValid() || !rcv_sock.IsValid())
	{
		cout << "Create socket failed." << endl;
		exit(1);
	}

	dpsd.start_time = DateTime::Now().GetTickCount();
	int64_t now = dpsd.start_time;
	
	if(!Thread::Create(rcv_thread, ThreadDelegate(this, &DVPinger::RecvProc), (void*)&dpsd))
	{
		cout << "Create receive thread failed." << endl;
		exit(1);
	}

	IPEndPoint ep(host, snd_port);
	int64_t send_interval = (int64_t)(packet_size * 0.5 / bandwidth);
	while(now - dpsd.start_time < period)
	{
		// send
		now = DateTime::Now().GetTickCount();
		int64_t expected_time = dpsd.start_time + (int64_t) (dpsd.dpr.bytes_sent_ideal / bandwidth);
		if(now < expected_time)
		{
			Thread::Sleep(expected_time - now);
		}
		if(snd_pkt_seq % dpsd.pkt_seq_divider == 0)
		{
			int64_t index = snd_pkt_seq / dpsd.pkt_seq_divider;
			if(index < 1000)
			{
				snd_pkt_time_cache[index] = DateTime::Now().GetTickCount();
			}
		}
		dpsd.dpr.bytes_sent_ideal += packet_size;
		int32_t bytes_sent = snd_sock.SendTo(packet, packet_size, ep);
		if(bytes_sent > 0)
			dpsd.dpr.bytes_sent += bytes_sent;
		++ dpsd.dpr.packets_sent;
		++ snd_pkt_seq;
		pRtpHdr->seq = htons(ntohs(pRtpHdr->seq) + 1);
	}
	dpsd.dpr.period = now - dpsd.start_time;

	QuitRecvThread();
	
	// receive as more extra packets as possible
	int64_t expire_time = 100000;
	if(dpsd.delay_hit_pkt_count == 0)
	{
		if(dpsd.first_recv_time != 0)
			expire_time = 4 * (dpsd.first_recv_time - dpsd.start_time);
	}
	else
		expire_time = 4 * dpsd.delay_sum / dpsd.delay_hit_pkt_count;
	if(expire_time < 100000)
		expire_time = 100000;
	if(expire_time > 1000000LL)
		expire_time = 1000000LL;
	int64_t last_recv_time = DateTime::Now().GetTickCount();
	now = last_recv_time;
	IPEndPoint rcvep;
	while(now - last_recv_time < expire_time)
	{
		int32_t bytes_recv = 0;
		while((bytes_recv = rcv_sock.ReceiveFrom(rcvbuf, packet_size, rcvep, 1000)) > 0)
		{
			if(rcvep.Address() != host)
				continue;
			now = DateTime::Now().GetTickCount();
			last_recv_time = now;
			if(rcv_pkt_seq % dpsd.pkt_seq_divider == 0)
			{
				int64_t index = rcv_pkt_seq / dpsd.pkt_seq_divider;
				if(index >= 0 && index < 1000 && snd_pkt_time_cache[index] > 0)
				{
					dpsd.delay_sum += now - snd_pkt_time_cache[index];
					++dpsd.delay_hit_pkt_count;
				}
			}
			++ dpsd.dpr.packets_recv;
			dpsd.dpr.bytes_recv += bytes_recv;
		} 
		now = DateTime::Now().GetTickCount();
	}
	if(dpsd.delay_hit_pkt_count == 0)
	{
		if(dpsd.first_recv_time != 0)
			dpsd.dpr.delay = (double)(dpsd.first_recv_time - dpsd.start_time);
		else
			dpsd.dpr.delay = -1.0;
	}
	else
	{
		dpsd.dpr.delay = (double)dpsd.delay_sum / dpsd.delay_hit_pkt_count;
	}

    return dpsd.dpr;
}

bool DVPinger::Cleanup()
{
	return snd_sock.Close() && rcv_sock.Close();
}


void DVPinger::QuitRecvThread()
{
	if(rcv_thread != NullThread)
	{
		bQuitRecv = true;
		evtRecvThreadQuit.WaitForSignal(1000000);
		if(rcv_thread != NullThread)
		{
			cout << "Receiving thread killed." << endl;
			Thread::Kill(rcv_thread);
			rcv_thread = NullThread;
		}
	}
	bQuitRecv = false;
	evtRecvThreadQuit.Reset();
}

